package org.robby.hbase.cdr;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;

public class ImportCdr {

	String tab_cdr_name = "tab_cdr";
	HTableInterface tab_cdr;
	Vector<Thread> thpool;

	public ImportCdr() {
		thpool = new Vector<Thread>();
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "192.168.1.90");
		try {
			HConnection hc = HConnectionManager.createConnection(conf);
			try {
				HBaseAdmin admin = new HBaseAdmin(hc);
				if (admin.tableExists(tab_cdr_name)) {
					admin.disableTable(tab_cdr_name);
					admin.deleteTable(tab_cdr_name);
				}
				HTableDescriptor tableDesc = new HTableDescriptor(tab_cdr_name);
				HColumnDescriptor hcolumn = new HColumnDescriptor("data");
				hcolumn.setMaxVersions(1);
				tableDesc.addFamily(hcolumn);
				admin.createTable(tableDesc);
				admin.close();
				tab_cdr = hc.getTable(tab_cdr_name);
			} catch (IOException e) {
				e.printStackTrace();
			}
		} catch (ZooKeeperConnectionException e) {
			e.printStackTrace();
		}
	}

	public void multi() throws Exception {
		String p = "/home/conkeyn/jar/output/";
		File path = new File(p);
		String[] list = path.list();
		int i = 0;
		Vector<String> vec = new Vector<String>();
		for (String s : list) {
			vec.add(p + s);
			i++;
			if (i % 2 == 0) {
				startThread(vec);
				vec = new Vector<String>();
			}
		}
		startThread(vec);
		while (!isAllThreadStopped()) {
			Thread.sleep(1000);
		}
	}

	public void single() throws Exception {
		// TODO Auto-generated method stub
		String p = "/home/conkeyn/jar/output/";
		File path = new File(p);
		String[] list = path.list();
		int i = 0;

		Vector<String> vec = new Vector<String>();
		for (String s : list) {
			vec.add(p + s);
		}
		startThread(vec);
		while (!this.isAllThreadStopped())
			Thread.sleep(1000);
	}

	public void startThread(Vector<String> vec) {
		System.out.println("Thread started");
		ImportThread t = new ImportThread(vec, tab_cdr);
		t.start();
		thpool.add(t);
	}

	public boolean isAllThreadStopped() {
		for (Thread t : thpool) {
			if (t.isAlive()) {
				return false;
			}
		}
		return true;
	}

	public class ImportThread extends Thread {
		Vector<String> files;
		HTableInterface tab_cdr;

		public ImportThread(Vector<String> files, HTableInterface tab_cdr) {
			super();
			this.files = files;
			this.tab_cdr = tab_cdr;
		}

		@Override
		public void run() {
			try {
				for (String f : files) {
					System.out.println(f);
					BufferedReader reader = new BufferedReader(new FileReader(new File(f)));
					String s;
					List<Put> ll = new ArrayList<Put>();
					int i = 0;
					while ((s = reader.readLine()) != null) {
						String[] arr = s.split(",");
						if (arr[0].equals(arr[2])) {
							continue;
						}
						String rowKey = arr[0] + "_" + arr[2] + "_" + arr[4];// 主叫+被叫+时间
						Put p = new Put(rowKey.getBytes());
						CdrPro.SmCdr cdr = CdrPro.SmCdr.newBuilder().setOaddr(arr[0]).setOareacode(arr[1]).setDaddr(arr[2]).setDareacode(arr[3])
								.setTimestamp(arr[4]).setType(arr[5]).build();
						p.add("data".getBytes(), null, cdr.toByteArray());
						// tab_cdr.put(p);
						ll.add(p);
						i++;
						if (i % 1000 == 0) {
							tab_cdr.put(ll);
							// tab_cdr.flushCommits();
							ll = new ArrayList<Put>();
						}

					}
					tab_cdr.put(ll);// 没有被100整除的余数
					// tab_cdr.flushCommits();
				}
				tab_cdr.close();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		ImportCdr importCdr = new ImportCdr();
		importCdr.single();
	}

}
